package com.peng.rpc.netty.client;

import com.peng.rpc.enumeration.RpcError;
import com.peng.rpc.exception.RpcException;
import com.peng.rpc.loadBalance.LoadBalance;
import com.peng.rpc.loadBalance.RandomLoadBalance;
import com.peng.rpc.loadBalance.RoundRobinLoadBalance;
import com.peng.rpc.netty.util.UnprocessedRequests;
import com.peng.rpc.registry.NacosServiceDiscovery;
import com.peng.rpc.registry.ServiceDiscovery;
import com.peng.rpc.serializer.CommonSerializer;
import com.peng.rpc.common.client.RpcClient;
import com.peng.rpc.entity.RpcRequest;
import com.peng.rpc.entity.RpcResponse;
import com.peng.rpc.util.ResponseChecker;
import com.peng.rpc.util.SingletonFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 利用netty传输协议包的类【实际是netty客户端的顶层入站】
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022/2/28 22:13
 */
@Slf4j
public class NettyRpcClient implements RpcClient {

    private static final Bootstrap BOOTSTRAP;
    private CommonSerializer serializer;
    private ServiceDiscovery serviceDiscovery;
    private static EventLoopGroup group;

    private UnprocessedRequests unprocessedRequests;

    public NettyRpcClient() {
        this(DEFAULT_SERIALIZER, new RoundRobinLoadBalance());
    }

    public NettyRpcClient(LoadBalance loadBalance) {
        this(DEFAULT_SERIALIZER, loadBalance);
    }

    public NettyRpcClient(int serializeCode) {
        this(serializeCode, new RandomLoadBalance());
    }

    public NettyRpcClient(int serializerCode, LoadBalance loadBalance) {
        this.serializer = CommonSerializer.getSerializerByCode(serializerCode);
        this.serviceDiscovery = new NacosServiceDiscovery(loadBalance);
        this.unprocessedRequests = SingletonFactory.getSingletonInstance(UnprocessedRequests.class);
    }

    static {
        BOOTSTRAP = new Bootstrap();
        group = new NioEventLoopGroup();
        BOOTSTRAP.group(group)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .channel(NioSocketChannel.class);
    }

    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            log.error("未设置序列化器，无法发送请求");
            throw new RpcException(RpcError.SERIALIZER_UNSETTED);
        }
        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        try {
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            Channel channel = ChannelProvider.getChannel(inetSocketAddress, serializer);
            if (channel != null && channel.isActive()) {
                log.info("客户端连接到服务器：【{}:{}】", inetSocketAddress.getAddress(), inetSocketAddress.getPort());
                unprocessedRequests.put(rpcRequest.getRequestId(), responseFuture);
                channel.writeAndFlush(rpcRequest)
                        .addListener((ChannelFutureListener) future -> {
                            if (future.isSuccess()) {
                                log.info("客户端成功发送消息：{}", rpcRequest);
//                                responseFuture.complete()
                            } else {
                                future.channel().close();
                                responseFuture.completeExceptionally(future.cause());
                                log.error("发送请求消息时发生错误：{}", future.cause().toString());
                            }
                        });

//                channel.closeFuture().sync();
//                //AttributeKey可以在一方的上下文对象中共享一个键值对，类似一个ThreadLocal。
//                //这里的RpcResponse对象是ClientHandler中设置进入的。
//                AttributeKey<RpcResponse> attributeKey = AttributeKey.valueOf("rpcResponse-" + rpcRequest.getRequestId());
//                RpcResponse rpcResponse = channel.attr(attributeKey).get();
//                //若响应包不合法，直接抛出异常。
//                ResponseChecker.check(rpcRequest, rpcResponse);
//                result.set(rpcResponse.getData());
            } else {
                group.shutdownGracefully();
                return null;
            }
        } catch (Exception e) {
            unprocessedRequests.remove(rpcRequest.getRequestId());
            Thread.currentThread().interrupt();
            log.error("客户端请求服务失败：{}", e.getMessage());
        }
        return responseFuture;
    }

}
